package npu_operator

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	corev1 "k8s.io/api/core/v1"
	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
	apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
	"k8s.io/kubernetes/test/e2e/framework"
)

var _ = SIGDescribe("NPU Operator has been installed", framework.WithLabel("npu_operator"), func() {
	var (
		clientset       *kubernetes.Clientset
		dynamicClient   dynamic.Interface
		extensionClient *apiextensionsclient.Clientset
		restConfig      *rest.Config
		ctx             context.Context
		err             error
	)

	BeforeEach(func() {
		ctx = context.Background()

		// 使用 kubeconfig 创建客户端
		kubeconfig := filepath.Join(homedir.HomeDir(), ".kube", "config")

		// 初始化 restConfig
		restConfig, err = clientcmd.BuildConfigFromFlags("", kubeconfig)
		Expect(err).NotTo(HaveOccurred())

		// 创建 clientset
		clientset, err = kubernetes.NewForConfig(restConfig)
		Expect(err).NotTo(HaveOccurred())

		// 创建动态客户端
		dynamicClient, err = dynamic.NewForConfig(restConfig)
		Expect(err).NotTo(HaveOccurred())

		// 创建扩展客户端
		extensionClient, err = apiextensionsclient.NewForConfig(restConfig)
		Expect(err).NotTo(HaveOccurred())
	})

	framework.Context("查找所有NPU Operator相关的Pod", framework.WithLabel("npu_operator_pods"), func() {
		var filteredPods []corev1.Pod

		BeforeEach(func() {
			filteredPods = []corev1.Pod{}
			// 查找所有NPU Operator相关的Pod
			filteredPods, err = filterNPUOperatorPods(clientset, ctx, filteredPods)
			Expect(err).NotTo(HaveOccurred(), "应该成功过滤出NPU Operator相关的Pod")
			// 至少找到一个npu-operator Pod
			Expect(len(filteredPods)).Should(BeNumerically(">", 0))
			GinkgoWriter.Printf("找到 %d 个包含'npu-operator'的Pod\n", len(filteredPods))
		})

		framework.It("643-查询对应NPUClusterPolicy状态", func() {
			// 获取NPUClusterPolicy
			npuPolicy, err, onlyOneNPUPolicy := getNPUPolicy(dynamicClient, ctx)
			Expect(err).NotTo(HaveOccurred(), "应该成功获取NPUClusterPolicy")
			Expect(onlyOneNPUPolicy).To(BeTrue(), "应该只有一个NPUClusterPolicy")

			// 输出NPUClusterPolicy对应信息
			policyName := npuPolicy.GetName()
			policyNamespace := npuPolicy.GetNamespace()
			GinkgoWriter.Printf("NPUClusterPolicy %s/%s\n", policyNamespace, policyName)

			// 查询NPUClusterPolicy对应状态
			state, err, found := getNPUClusterPolicyStatus(npuPolicy)
			Expect(err).NotTo(HaveOccurred(), "应该成功获取NPUClusterPolicy对应状态")
			Expect(found).To(BeTrue(), "应该能够找i到NPUClusterPolicy的状态字段")

			GinkgoWriter.Printf("该NPUClusterPolicy的状态: %s\n", state)

			// 验证Pod状态为running
			podRunning := isPodsRunning(filteredPods)

			if podRunning {
				// 检查状态是否为 ready
				GinkgoWriter.Printf("  - 命名空间: %s, NPUClusterPolicy: %s, 状态: %s\n",
					policyNamespace, policyName, state)

				Expect(strings.ToLower(state)).To(Equal("ready"), "NPUClusterPolicy的状态应该为Ready")
			} else {
				// 检查状态是否为 notReady
				GinkgoWriter.Printf("  - 命名空间: %s, NPUClusterPolicy: %s, 状态: %s\n",
					policyNamespace, policyName, state)

				Expect(strings.ToLower(state)).To(Equal("notready"), "NPUClusterPolicy的状态应该为notReady")
			}
		})

		// 验证Ascend Driver & Firmware组件
		verifyAscendDriverAndFrimware := func() {
			// 执行 npu-smi info 命令
			cmd := exec.Command("npu-smi", "info")
			output, err := cmd.CombinedOutput()

			// 验证命令执行无错误且输出不为空
			Expect(err).NotTo(HaveOccurred(), "'npu-smi info'命令应执行成功")
			Expect(string(output)).NotTo(BeEmpty(), "'npu-smi info'命令的输出结果不应为空")
			GinkgoWriter.Printf("npu-smi info 的执行结果不为空，Ascend Driver & Firmware组件验证成功\n")
		}

		// 验证Ascend Docker Runtime组件
		verifyAscendDockerRuntime := func() {
			// 执行 ls /dev 命令
			cmd := exec.Command("ls", "/dev")
			output, err := cmd.CombinedOutput()

			// 验证命令执行无错误
			Expect(err).NotTo(HaveOccurred(), "'ls /dev'命令应执行成功")

			// 将输出转换为字符串
			outputStr := string(output)

			// 验证输出中是否包含 "davinci0"
			Expect(outputStr).To(ContainSubstring("davinci0"),
				"'ls /dev'命令的输出结果应该包含'davinci0'")
			GinkgoWriter.Printf("ls /dev 的执行结果中包含davinci0，Ascend Docker Runtime组件验证成功\n")
		}

		// 验证Ascend Device Plugin组件
		verifyAscendDevicePlugin := func() {
			// 执行kubectl describe node
			cmd := exec.Command("kubectl", "describe", "node")
			output, err := cmd.CombinedOutput()

			// 验证命令执行成功
			Expect(err).NotTo(HaveOccurred(), "'kubectl describe node'命令应执行成功")

			outputStr := string(output)

			// 检查输出中是否包含Capacity部分
			Expect(outputStr).To(ContainSubstring("Capacity"),
				"输出应该包含'Capacity'部分")

			// 提取Capacity字段中的NPU信息
			npuCount, err, foundNPU := extractNPUFromCapacity(outputStr)

			// 验证是否找到NPU
			Expect(err).NotTo(HaveOccurred(), "应该成功提取Capacity字段中的NPU信息")
			Expect(foundNPU).To(BeTrue(), "应该成功找到'NPU'部分字段")
			Expect(npuCount).To(BeNumerically(">", 0),
				"在Capacity部分应找到有效的huawei.com/Ascend字段")

			GinkgoWriter.Printf("Capacity部分包含 %d 个 Ascend NPU，Ascend Device Plugin组件验证成功\n", npuCount)
		}

		// 通过 kubectl logs pod 验证组件
		verifyPodLogs := func(podName string, keyWords []string) {
			// 获取对应的Pod
			targetPodList := findTargetPodList(podName, filteredPods)
			Expect(len(targetPodList)).NotTo(Equal(0), "应该在filtered pods集合中找到"+podName+" pods")

			for _, targetPod := range targetPodList {
				// 获取日志
				logContent, err := getPodLogs(clientset, targetPod)
				Expect(err).NotTo(HaveOccurred(), "应成功获取"+podName+" pod的日志")
				Expect(logContent).NotTo(BeEmpty(), podName+" pod的日志内容不应为空")

				// 检查日志中是否包含关键字
				for _, keyWord := range keyWords {
					Expect(logContent).To(ContainSubstring(keyWord),
						podName+"日志应该包含'"+keyWord+"'")
					GinkgoWriter.Printf(podName + "日志包含关键字:" + keyWord + "\n")
				}
			}

			GinkgoWriter.Printf("Pod logs 验证结束\n")
		}

		// 验证Ascend Operator组件
		verifyAscendOperator := func() {
			podName := "ascend-operator-manager"
			keyWords := []string{"init success", "starting manager"}
			verifyPodLogs(podName, keyWords)

			// 打印确认信息
			GinkgoWriter.Printf(podName + "日志包含所有需要的关键字，Ascend Operator组件验证成功\n")
		}

		// 验证NodeD组件
		verifyNodeD := func() {
			podName := "noded"
			keyWords := []string{"init success", "init config from local json file success"}
			verifyPodLogs(podName, keyWords)

			// 打印确认信息
			GinkgoWriter.Printf(podName + "日志包含所有需要的关键字，NodeD组件验证成功\n")
		}

		// 验证ClusterD组件
		verifyClusterD := func() {
			podName := "clusterd"
			keyWords := []string{"init success", "start listen"}
			verifyPodLogs(podName, keyWords)

			// 打印确认信息
			GinkgoWriter.Printf(podName + "日志包含所有需要的关键字，ClusterD组件验证成功\n")
		}

		// 验证NPU Exporter组件
		verifyNPUExporter := func() {
			podName := "npu-exporter"
			keyWords := []string{"update Cache,key is npu-exporter"}
			verifyPodLogs(podName, keyWords)

			// 打印确认信息
			GinkgoWriter.Printf(podName + "日志包含所有需要的关键字，NPU Exporter组件验证成功\n")
		}

		// 验证MindIO组件
		verifyMindIO := func() {
			podName := "mindio"
			keyWords := []string{"Successfully installed"}
			verifyPodLogs(podName, keyWords)

			// 打印确认信息
			GinkgoWriter.Printf(podName + "日志包含所有需要的关键字，MindIO组件验证成功\n")
		}

		// 验证Volcano组件
		verifyVolcano := func() {
			// 验证有volcano-controllers
			podName := "volcano-controllers"
			targetPodList := findTargetPodList(podName, filteredPods)
			Expect(len(targetPodList)).NotTo(Equal(0), "应该在filtered pods集合中找到volcano-controllers pods")

			// 查看volcano-controllers日志
			logFilePath := "/var/log/mindx-dl/volcano-controller/volcano-controller.log"
			logContent, err := os.ReadFile(logFilePath)
			Expect(err).NotTo(HaveOccurred(), "读取日志文件失败")

			// 检查特定字段
			Expect(string(logContent)).To(ContainSubstring("running"), "volcano-controllers日志中应包含'running'")
			Expect(string(logContent)).To(ContainSubstring("start"), "volcano-controllers日志中应包含'start'")
			GinkgoWriter.Printf("volcano-controllers日志中包含'running'和'start'字段，Volcano组件验证成功\n")
		}

		framework.It("151-各组件验证", func() {
			verifyAscendDriverAndFrimware() // 验证Ascend Driver & Firmware组件
			verifyAscendDockerRuntime()     // 验证Ascend Docker Runtime组件
			verifyAscendDevicePlugin()      // 验证Ascend Device Plugin组件
			verifyAscendOperator()          // 验证Ascend Operator组件
			verifyNodeD()                   // 验证NodeD组件
			verifyClusterD()                // 验证ClusterD组件
			verifyNPUExporter()             // 验证NPU Exporter组件
			verifyMindIO()                  // 验证MindIO组件
			verifyVolcano()                 // 验证Volcano组件
		})
	})

	Context("查询所有的节点", framework.WithLabel("npu_operator_nodes"), func() {
		var nodes *corev1.NodeList

		BeforeEach(func() {
			nodes, err = clientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{})

			Expect(err).NotTo(HaveOccurred(), "'kubectl get nodes' 执行失败")
			Expect(nodes.Items).NotTo(BeEmpty(), "在集群中没有找到任何节点")

			GinkgoWriter.Printf("检测到nodes数量:\n%d\n", len(nodes.Items))
		})

		// 验证节点标签是否符合预期
		verifyNodeLabels := func(node corev1.Node, label string, expectValue string) {
			value, exists := node.Labels[label]

			Expect(exists).To(BeTrue(), label+"标签应当存在")
			Expect(value).To(Equal(expectValue), label+"标签的值应为"+expectValue)
			GinkgoWriter.Printf("%s节点存在%s标签，值为%s\n", node.Name, label, value)
		}

		framework.It("669-检查master和worker节点对应的标签（多节点）", func() {
			for _, node := range nodes.Items {
				// 检测是否为master节点
				_, controlPlaneExists := node.Labels["node-role.kubernetes.io/control-plane"]
				_, masterExists := node.Labels["node-role.kubernetes.io/master"]
				masterLabelExists := controlPlaneExists && masterExists

				if masterLabelExists {
					// master节点
					GinkgoWriter.Printf("%s节点是master节点\n", node.Name)
					label := "masterselector"
					expectValue := "dls-master-node"

					// 检查master节点标签
					verifyNodeLabels(node, label, expectValue)
				} else {
					// worker节点
					GinkgoWriter.Printf("%s节点是worker节点\n", node.Name)
					_, exists := node.Labels["masterselector"]

					Expect(exists).To(BeFalse(), "worker节点不应存在master节点标签")
					GinkgoWriter.Printf("%s节点不存在masterselector标签\n", node.Name)
				}
			}
		})

		framework.It("670-检查容器运行时标签", func() {
			label := "openfuyao.com/container.runtime"
			expectValue := "containerd"

			// 检查每个节点的标签
			for _, node := range nodes.Items {
				verifyNodeLabels(node, label, expectValue)
			}
		})

		framework.It("679-检查新增NFD服务发现标签（910-D802）", func() {
			label := "feature.node.kubernetes.io/pci-1200_19e5_d802.present"
			expectValue := "true"

			// 检查每个节点的标签
			for _, node := range nodes.Items {
				verifyNodeLabels(node, label, expectValue)
			}
		})

		framework.It("680-检查节点NPU设备标签", func() {
			label := "openfuyao.com/npu.present"
			expectValue := ""

			// 检查每个节点的标签
			for _, node := range nodes.Items {
				verifyNodeLabels(node, label, expectValue)
			}
		})

		framework.It("681-检查管理节点标签（单节点）", func() {
			for _, node := range nodes.Items {
				_, controlPlaneExists := node.Labels["node-role.kubernetes.io/control-plane"]
				_, masterExists := node.Labels["node-role.kubernetes.io/master"]
				masterLabelExists := controlPlaneExists && masterExists

				if masterLabelExists {
					// master节点
					GinkgoWriter.Printf("%s节点是master节点\n", node.Name)
					label := "masterselector"
					expectValue := "dls-master-node"

					// 检查master节点标签
					verifyNodeLabels(node, label, expectValue)
				} else {
					// worker节点
					GinkgoWriter.Printf("%s节点是worker节点，跳过验证\n", node.Name)
					continue
				}
			}
		})

		framework.It("682-检查NPU-operator节点标签（Atalas 800I-Ascend 910B）ARM", func() {
			for _, node := range nodes.Items {
				servertype, exists := node.Labels["servertype"]
				Expect(exists).To(BeTrue())

				if servertype == "Ascend910B-20" {
					GinkgoWriter.Printf("%s节点存在servertype，值为%s，符合环境要求，继续验证\n", node.Name, servertype)
					label := "node-role.kubernetes.io/worker"
					expectValue := "worker"

					// 检查节点标签
					verifyNodeLabels(node, label, expectValue)
				} else {
					GinkgoWriter.Printf("%s节点存在servertype，值为%s，不符合环境要求，跳过验证\n", node.Name, servertype)
					continue
				}
			}
		})
	})

	framework.Context("查询所有的服务", framework.WithLabel("npu_operator_services"), func() {
		var services *corev1.ServiceList

		BeforeEach(func() {
			services, err = clientset.CoreV1().Services("").List(ctx, metav1.ListOptions{})

			Expect(err).NotTo(HaveOccurred(), "'kubectl get svc -A' 应执行成功")
			Expect(services.Items).NotTo(BeEmpty(), "集群上应发现服务")

			GinkgoWriter.Printf("检测到services数量:\n%d\n", len(services.Items))
		})

		// 验证Pod运行状态
		verifyPodRunning := func(podName string) {
			GinkgoWriter.Printf("等待 Pod：%s 运行...\n", podName)

			Eventually(func() (corev1.PodPhase, error) {
				currentPod, err := clientset.CoreV1().Pods("default").Get(ctx, podName, metav1.GetOptions{})
				if err != nil {
					GinkgoWriter.Printf("获取 Pod：%s 失败: %v\n", podName, err)
					return "", err
				}

				GinkgoWriter.Printf("当前 Pod：%s 状态: %s\n", currentPod.Name, currentPod.Status.Phase)
				return currentPod.Status.Phase, nil
			}).WithTimeout(60*time.Second).WithPolling(2*time.Second).Should(Equal(corev1.PodRunning),
				"Pod：%s 未在指定时间内运行", podName)
		}

		// 通过url查看指标
		verifyMetricsThroughURL := func(metricsUrl string, podName string, containerName string) {
			var metricsResponse string
			GinkgoWriter.Printf("开始获取指标数据，URL: %s\n", metricsUrl)

			Eventually(func() bool {
				// 创建带超时的 HTTP 客户端
				client := &http.Client{
					Timeout: 30 * time.Second,
				}

				// 发送请求
				resp, err := client.Get(metricsUrl)
				if err != nil {
					GinkgoWriter.Printf("访问指标端点失败: %v\n", err)
					return false
				}
				defer func() {
					if err := resp.Body.Close(); err != nil {
						GinkgoWriter.Printf("关闭响应体失败: %v\n", err)
					}
				}()

				// 检查状态码
				if resp.StatusCode != http.StatusOK {
					GinkgoWriter.Printf("指标端点返回非200状态码: %d\n", resp.StatusCode)
					return false
				}

				// 读取响应
				body, err := io.ReadAll(resp.Body)
				if err != nil {
					GinkgoWriter.Printf("读取响应体失败: %v\n", err)
					return false
				}

				// 判断指标是否符合要求
				metricsResponse = string(body)
				identification := identifyMetrics(metricsResponse, podName, containerName)
				return identification
			}).WithTimeout(10*time.Second).WithPolling(1*time.Second).Should(BeTrue(),
				"在重试10次后仍未能成功获取有效的指标数据")
		}

		// 验证 Pod 已被删除
		verifyNPUOperatorTestPodDeleted := func(podName string) {
			// 验证 Pod 已被删除
			Eventually(func() (bool, error) {
				currentPod, err := clientset.CoreV1().Pods("default").Get(ctx, podName, metav1.GetOptions{})
				if currentPod != nil && err == nil {
					GinkgoWriter.Printf("当前 Pod：%s 状态: %s\n", currentPod.Name, currentPod.Status.Phase)
				}
				if errors.IsNotFound(err) {
					return errors.IsNotFound(err), nil
				} else if err != nil {
					return false, fmt.Errorf("获取 Pod：%s 失败： %v\n", podName, err)
				}
				return false, nil
			}, 60*time.Second, 2*time.Second).Should(BeTrue(), "Pod应该被删除")
		}

		framework.It("650-NPU Exporter组件CRD文件字段验证", func() {
			//查找NPU Exporter服务
			clusterIP, servicePort, found := findNPUExporterService(services)

			// 首先检查是否找到了服务
			Expect(found).To(BeTrue(), "未找到 npu-exporter/npu-exporter-service 服务")

			// 然后检查 ClusterIP 和 servicePort 是否有效（不是空且不是 "None"）
			Expect(clusterIP).NotTo(BeEmpty(), "服务的 ClusterIP 为空")
			Expect(clusterIP).NotTo(Equal("None"), "服务是 Headless Service，没有 ClusterIP")
			Expect(servicePort).NotTo(BeZero(), "服务没有定义端口")

			GinkgoWriter.Printf("找到服务: ClusterIP=%s, Port=%d\n", clusterIP, servicePort)

			// 创建测试用的Pod
			podName, containerName, err := createNPUOperatorTestPod(clientset, ctx)
			Expect(err).NotTo(HaveOccurred(), "应成功创建npu-operator-test-pod")

			// 验证Pod运行状态
			verifyPodRunning(podName)
			GinkgoWriter.Printf("Pod：%s 正在运行!\n", podName)

			// 通过url查看指标
			var metricsUrl = "http://" + clusterIP + ":" + fmt.Sprintf("%d", servicePort) + "/metrics"
			verifyMetricsThroughURL(metricsUrl, podName, containerName)

			GinkgoWriter.Printf("成功验证 container_name 为 %s， pod_name 为 %s\n", containerName, podName)

			// 删除测试用的Pod，恢复原来的环境
			err = deleteNPUOperatorTestPod(clientset, ctx, podName)
			Expect(err).NotTo(HaveOccurred(), "'kubectl delete npu-operator pod -n "+podName+"' 应执行成功")
			// 验证Pod已删除
			verifyNPUOperatorTestPodDeleted(podName)
			GinkgoWriter.Printf("npu-operator-test-pod已删除，原有环境已恢复")
		})
	})

	Context("查找NPU Operator相关的CR", framework.WithLabel("npu_operator_cr"), func() {
		var (
			NPUOperatorCRD apiextensionsv1.CustomResourceDefinition
			NPUOperatorCR  unstructured.Unstructured
			gvr            schema.GroupVersionResource
		)

		BeforeEach(func() {
			// 获取环境中CRD列表
			crds, err := extensionClient.ApiextensionsV1().CustomResourceDefinitions().List(context.TODO(), metav1.ListOptions{})
			Expect(err).NotTo(HaveOccurred(), "应该成功获取CRD列表")
			Expect(len(crds.Items)).Should(BeNumerically(">=", 1), "CRD数量应为至少1个")
			GinkgoWriter.Printf("找到%d个CRD\n", len(crds.Items))

			// 过滤出NPU Operator相关的CRD
			NPUOperatorCRDs, found := filterNPUOperatorCRDs(crds)
			Expect(found).To(BeTrue(), "应该能够找到NPU Operator CRD")
			Expect(len(NPUOperatorCRDs)).Should(BeNumerically("==", 1), "NPU Operator CRD数量应为1个")
			GinkgoWriter.Printf("成功找到NPU Operator CRD\n")

			NPUOperatorCRD = NPUOperatorCRDs[0]

			// 定义gvr结构
			gvr = schema.GroupVersionResource{
				Group:    NPUOperatorCRD.Spec.Group,
				Version:  NPUOperatorCRD.Spec.Versions[0].Name, // 使用第一个可用版本
				Resource: NPUOperatorCRD.Spec.Names.Plural,
			}

			// 找到该CRD对应的CR
			currentNPUOperatorCR, err := getNPUOperatorCR(dynamicClient, gvr)
			Expect(err).NotTo(HaveOccurred(), "应该成功获取最新NPUOperatorCR")
			NPUOperatorCR = currentNPUOperatorCR

			// 检验当前CR状态
			isReady, err := isNPUOperatorCRReady(currentNPUOperatorCR)
			Expect(err).NotTo(HaveOccurred(), "获取当前CR状态失败")
			Expect(isReady).To(BeTrue(), "NPUOperatorCR的状态应为Ready")

			GinkgoWriter.Printf("NPUOperatorCR：%s 验证成功，状态为Ready\n", NPUOperatorCR.GetName())
		})

		// 执行修改操作并验证修改有效性
		verifyModificationEffectiveness := func(componentName string, managed bool, expectReason string, expectType string) {
			// 修改 Spec.componentName.Managed 为 managed
			NPUOperatorCR, err = modifyComponentManaged(componentName, managed, NPUOperatorCR, dynamicClient, ctx, gvr)
			Expect(err).NotTo(HaveOccurred(), "应成功将"+componentName+"的Managed值修改为"+strconv.FormatBool(managed))
			GinkgoWriter.Printf("成功将%s的Managed值修改为%t\n", componentName, managed)

			var finalReason, finalCurrentType string

			// 验证修改后结果是否符合预期
			Eventually(func() bool {
				// 获取最新的CR
				currentNPUOperatorCR, err := getNPUOperatorCR(dynamicClient, gvr)
				if err != nil {
					return false
				}
				isReady, err := isNPUOperatorCRReady(currentNPUOperatorCR)
				if err != nil || !isReady {
					return false
				}
				NPUOperatorCR = currentNPUOperatorCR

				// 获取CR状态
				reason, currentType, err := getComponentState(componentName, NPUOperatorCR)
				if err != nil {
					GinkgoWriter.Printf("获取状态失败: %v\n", err)
					return false
				}

				// 验证CR状态是否符合预期
				finalReason = reason
				finalCurrentType = currentType

				success := reason == expectReason && currentType == expectType
				if !success {
					GinkgoWriter.Printf("状态不匹配: currentReason=%s, currentType=%s\n", reason, currentType)
				}

				return success
			}, 15*time.Second, 1*time.Second).Should(BeTrue(),
				"组件状态应该在15秒内达到预期。最终状态: reason="+finalReason+", currentType="+finalCurrentType)
			GinkgoWriter.Printf("状态匹配: reason=%s, currentType=%s\n", finalReason, finalCurrentType)
		}

		// 通过修改Managed值验证组件
		verifyComponent := func(componentName string) {
			// 验证修改 Spec.componentName.Managed 为 false
			var managed = false
			var expectReason = "ComponentUnmanaged"
			var expectType = "unmanaged"
			verifyModificationEffectiveness(componentName, managed, expectReason, expectType)

			// 设置回true，恢复原来的环境
			managed = true
			expectReason = "Reconciled"
			expectType = "running"
			verifyModificationEffectiveness(componentName, managed, expectReason, expectType)

			GinkgoWriter.Printf("%s组件验证成功", componentName)
		}

		framework.It("609-MindIO组件状态验证", func() {
			var componentName = "mindioacp"
			verifyComponent(componentName) // 验证Mindioacp组件

			componentName = "mindiotft"
			verifyComponent(componentName) // 验证Mindiotft组件
		})
	})

})

// 以下为辅助函数

// 查找所有NPU Operator相关的Pod
func filterNPUOperatorPods(clientset *kubernetes.Clientset, ctx context.Context, filteredPods []corev1.Pod) ([]corev1.Pod, error) {
	// 获取所有命名空间的Pod
	podList, err := clientset.CoreV1().Pods("").List(ctx, metav1.ListOptions{})
	if err != nil {
		return nil, fmt.Errorf("列举pods失败: %w", err)
	}

	// 过滤npu_operator相关的Pod
	var prefixes = []string{
		"ascend-runtime-containerd",
		"npu-driver",
		"npu-operator",
		"npu-feature-discovery",
		"mindio-acp",
		"mindio-tft",
		"ascend-device-plugin",
		"ascend-operator-manager",
		"clusterd",
		"noded",
		"resilience-controller",
		"npu-exporter",
		"volcano-controllers",
		"volcano-scheduler",
	}
	for _, pod := range podList.Items {
		for _, prefix := range prefixes {
			if strings.HasPrefix(pod.Name, prefix) {
				filteredPods = append(filteredPods, pod)
				break
			}
		}
	}

	return filteredPods, nil
}

// 验证Pod状态为running
func isPodsRunning(filteredPods []corev1.Pod) bool {
	podRunning := true
	for _, pod := range filteredPods {
		GinkgoWriter.Printf("  - 命名空间: %s, Pod名称: %s, 状态: %s\n",
			pod.Namespace, pod.Name, pod.Status.Phase)
		if pod.Status.Phase != corev1.PodRunning {
			podRunning = false
			GinkgoWriter.Printf("Pod %s 状态非running: %s\n", pod.Name, pod.Status.Phase)
		}
	}
	return podRunning
}

// 获取NPUClusterPolicy
func getNPUPolicy(dynamicClient dynamic.Interface, ctx context.Context) (unstructured.Unstructured, error, bool) {
	// 定义 NPUClusterPolicy 的 GVR (Group Version Resource)
	// 注意：需要根据实际的 CRD 定义调整这些值
	gvr := schema.GroupVersionResource{
		Group:    "npu.openfuyao.com",  // 根据实际的CRD组名调整
		Version:  "v1",                 // 根据实际的版本调整
		Resource: "npuclusterpolicies", // 复数形式
	}

	// 获取所有命名空间的 NPUClusterPolicy
	npuPolicies, err := dynamicClient.Resource(gvr).Namespace("").List(ctx, metav1.ListOptions{})
	if err != nil {
		return unstructured.Unstructured{}, fmt.Errorf("获取NPUClusterPolicies失败: %w", err), false
	}

	GinkgoWriter.Printf("找到 %d 个 NPUClusterPolicy 资源：\n", len(npuPolicies.Items))

	// 验证只有一个 NPUClusterPolicy
	if len(npuPolicies.Items) != 1 {
		GinkgoWriter.Printf("NPUClusterPolicy数量不为1，不符合要求")
		return unstructured.Unstructured{}, nil, false
	}
	return npuPolicies.Items[0], nil, true
}

// 获取NPUClusterPolicy的状态字段，具体路径根据实际的 CRD 结构调整（本用例为Status.Conditions[0].Type）
func getNPUClusterPolicyStatus(npuPolicy unstructured.Unstructured) (string, error, bool) {
	// 获取Status字段
	status, found, err := unstructured.NestedMap(npuPolicy.Object, "status")
	if err != nil {
		return "", fmt.Errorf("获取NPUClusterPolicy Status失败: %w", err), false
	}
	if !found {
		GinkgoWriter.Printf("未找到 Status 或 Status 为空\n")
		return "", nil, false
	}

	// 获取Conditions字段
	conditions, found, err := unstructured.NestedSlice(status, "conditions")
	if err != nil {
		return "", fmt.Errorf("获取NPUClusterPolicy Conditions失败: %w", err), false
	}
	if !found || len(conditions) == 0 {
		GinkgoWriter.Printf("未找到 conditions 或 conditions 为空\n")
		return "", nil, false
	}

	// 获取第一个condition
	firstCondition, ok := conditions[0].(map[string]interface{})
	if !ok {
		GinkgoWriter.Printf("第一个 condition 格式错误\n")
		return "", nil, false
	}

	// 最终需要识别的状态在Status.Conditions[0].Type中
	state, found, err := unstructured.NestedString(firstCondition, "type")
	if err != nil {
		return "", fmt.Errorf("获取NPUClusterPolicy Type失败: %w", err), false
	}
	if !found {
		GinkgoWriter.Printf("未找到 Type 或 Type 为空\n")
		return "", nil, false
	}

	return state, nil, true
}

// 获取Pod日志内容
func getPodLogs(clientset *kubernetes.Clientset, targetPod *corev1.Pod) (string, error) {
	// 获取Pod日志
	podLogOpts := corev1.PodLogOptions{}
	req := clientset.CoreV1().Pods(targetPod.Namespace).GetLogs(targetPod.Name, &podLogOpts)

	podLogs, err := req.Stream(context.TODO())
	if err != nil {
		return "", fmt.Errorf("打开log stream失败: %w", err)
	}
	defer func(podLogs io.ReadCloser) {
		err := podLogs.Close()
		if err != nil {
			GinkgoWriter.Printf("关闭pod logs失败: %v\n", err)
		}
	}(podLogs)

	// 读取日志内容
	logBuf := new(bytes.Buffer)
	_, err = io.Copy(logBuf, podLogs)
	if err != nil {
		return "", fmt.Errorf("读取log content失败: %w", err)
	}

	return logBuf.String(), nil
}

// 提取Capacity字段中的NPU信息
func extractNPUFromCapacity(outputStr string) (int, error, bool) {
	// 提取Capacity部分
	lines := strings.Split(outputStr, "\n")
	inCapacity := false
	var npuCount int
	foundNPU := false

	for _, line := range lines {
		line = strings.TrimSpace(line)

		// 进入Capacity部分
		if strings.Contains(line, "Capacity:") {
			inCapacity = true
			continue
		}

		// 离开Capacity部分
		if inCapacity && (line == "Allocatable:" || line == "") {
			inCapacity = false
			break
		}

		// 在Capacity部分查找NPU
		if inCapacity && strings.Contains(line, "huawei.com/Ascend") {
			foundNPU = true
			GinkgoWriter.Printf("找到NPU字段：%s\n", line)
			parts := strings.Split(line, ":")
			if len(parts) != 2 {
				return 0, fmt.Errorf("NPU字段格式错误: %s", line), false
			}

			countStr := strings.TrimSpace(parts[1])
			count, err := strconv.Atoi(countStr)
			if err != nil {
				return 0, fmt.Errorf("NPU数量解析失败: %w", err), false
			}

			npuCount = count
			break
		}
	}
	return npuCount, nil, foundNPU
}

// 获取前缀为podName的Pods
func findTargetPodList(podName string, filteredPods []corev1.Pod) []*corev1.Pod {
	var targetPodList []*corev1.Pod
	for i, pod := range filteredPods {
		if strings.HasPrefix(pod.Name, podName) {
			targetPod := &filteredPods[i]
			GinkgoWriter.Printf("找到target pod: %s in namespace: %s\n",
				targetPod.Name, targetPod.Namespace)
			targetPodList = append(targetPodList, targetPod)
		}
	}
	return targetPodList
}

// 查找NPU Exporter服务
func findNPUExporterService(services *corev1.ServiceList) (string, int32, bool) {
	var found = false
	for _, service := range services.Items {
		if service.Namespace == "npu-exporter" && service.Name == "npu-exporter-service" {
			found = true
			clusterIP := service.Spec.ClusterIP
			ports := service.Spec.Ports
			if len(ports) == 0 {
				GinkgoWriter.Printf("NPU Exporter服务端口号为空，不符合规范")
				return "", 0, found
			}

			servicePort := ports[0].Port
			return clusterIP, servicePort, found
		}
	}
	return "", 0, found
}

// 创建测试Pod对象
func createPod() *corev1.Pod {
	// 定义NPU资源
	npuResource := resource.MustParse("1")

	// 创建Pod对象
	pod := &corev1.Pod{
		TypeMeta: metav1.TypeMeta{
			APIVersion: "v1",
			Kind:       "Pod",
		},
		ObjectMeta: metav1.ObjectMeta{
			Name: "npu-operator-test-pod",
		},
		Spec: corev1.PodSpec{
			Containers: []corev1.Container{
				{
					Name:            "npu-operator-test-container",
					Image:           "docker.io/rayproject/ray",
					ImagePullPolicy: corev1.PullIfNotPresent,
					Command: []string{
						"/bin/bash",
						"-c",
						"while true;do sleep 30;done",
					},
					Resources: corev1.ResourceRequirements{
						Requests: corev1.ResourceList{
							"huawei.com/Ascend310P": npuResource,
						},
						Limits: corev1.ResourceList{
							"huawei.com/Ascend310P": npuResource,
						},
					},
				},
			},
		},
	}

	GinkgoWriter.Printf("npu-operator-test-pod Pod对象创建成功\n")
	return pod
}

// 创建 npu-operator-test-pod
func createNPUOperatorTestPod(clientset *kubernetes.Clientset, ctx context.Context) (string, string, error) {
	// 解析文件
	npuOperatorTestPod := createPod()

	// 提取 Pod 名称
	podName := npuOperatorTestPod.ObjectMeta.Name
	GinkgoWriter.Printf("Pod名称: %s\n", podName)

	// 提取所有容器名称
	containerName := npuOperatorTestPod.Spec.Containers[0].Name
	GinkgoWriter.Printf("Container名称: %s\n", containerName)

	// 创建使用npu的pod
	GinkgoWriter.Printf("开始创建 Pod： %s ，Container: %s\n", podName, containerName)
	_, err := clientset.CoreV1().Pods("default").Create(ctx, npuOperatorTestPod, metav1.CreateOptions{})
	if err != nil {
		return "", "", fmt.Errorf("创建 Pod 失败: %w", err)
	}

	return podName, containerName, nil
}

// 判断指标是否符合要求
func identifyMetrics(metricsResponse string, podName string, containerName string) bool {
	// 验证指标数据不为空
	if metricsResponse == "" {
		GinkgoWriter.Printf("指标数据为空\n")
		return false
	}

	// 验证指标数据中是否包含正确的 container_name 和 pod_name
	hasContainerName := strings.Contains(metricsResponse, "container_name=\""+containerName+"\"")
	hasPodName := strings.Contains(metricsResponse, "pod_name=\""+podName+"\"")

	if !hasContainerName || !hasPodName {
		GinkgoWriter.Printf("指标数据验证失败 - 包含 container_name: %t, 包含 pod_name: %t\n",
			hasContainerName, hasPodName)
		return false
	}

	GinkgoWriter.Printf("成功获取并验证指标数据，数据长度: %d 字节\n", len(metricsResponse))
	return true
}

// 删除 npu-operator-test-pod
func deleteNPUOperatorTestPod(clientset *kubernetes.Clientset, ctx context.Context, podName string) error {
	// 删除 Pod
	GinkgoWriter.Printf("开始删除测试Pod，恢复原有环境：\n")
	err := clientset.CoreV1().Pods("default").Delete(ctx, podName, metav1.DeleteOptions{})
	if err != nil {
		return fmt.Errorf("删除 Pod 失败: %w", err)
	}
	return nil
}

// 过滤包含 "npuclusterpolicies" 且状态正常的 CRD
func filterNPUOperatorCRDs(crds *apiextensionsv1.CustomResourceDefinitionList) ([]apiextensionsv1.CustomResourceDefinition, bool) {
	GinkgoWriter.Printf("%-50s %-10s\n", "NAME", "ESTABLISHED")
	var NPUOperatorCRDs []apiextensionsv1.CustomResourceDefinition
	found := false
	for _, crd := range crds.Items {
		// 过滤名字包含"npuclusterpolicies"
		key := "npuclusterpolicies"
		if strings.Contains(strings.ToLower(crd.Name), key) {
			established := "False"
			// 判断 CRD 是否已成功注册到 API server 并可以正常使用
			for _, condition := range crd.Status.Conditions {
				if condition.Type == apiextensionsv1.Established {
					if condition.Status == apiextensionsv1.ConditionTrue {
						established = "True"
						NPUOperatorCRDs = append(NPUOperatorCRDs, crd)
						found = true
					}
					break
				}
			}
			GinkgoWriter.Printf("%-50s %-10s\n", crd.Name, established)
		}
	}
	return NPUOperatorCRDs, found
}

func getNPUOperatorCR(dynamicClient dynamic.Interface, gvr schema.GroupVersionResource) (unstructured.Unstructured, error) {
	// 找到该CRD对应的CR（取第一个，其余的无作用）
	crList, err := dynamicClient.Resource(gvr).Namespace("").List(context.TODO(), metav1.ListOptions{})
	if err != nil {
		return unstructured.Unstructured{}, fmt.Errorf("获取CR列表失败: %w", err)
	}
	if len(crList.Items) == 0 {
		return unstructured.Unstructured{}, fmt.Errorf("CR数量应为至少1个")
	}

	// 取第一个CR实例
	NPUOperatorCR := crList.Items[0]
	GinkgoWriter.Printf("找到CR: %s\n", NPUOperatorCR.GetName())

	return NPUOperatorCR, nil
}

// 检查NPUOperatorCR状态是否为Ready
func isNPUOperatorCRReady(NPUOperatorCR unstructured.Unstructured) (bool, error) {
	// 获取当前CR状态
	state, err, found := getNPUClusterPolicyStatus(NPUOperatorCR)
	if err != nil {
		return false, fmt.Errorf("获取NPUClusterPolicy CR状态失败: %w", err)
	}
	if !found {
		return false, fmt.Errorf("未找到NPUClusterPolicy CR的状态字段")
	}

	// 检查CR状态是否为Ready
	GinkgoWriter.Printf("该NPUClusterPolicy的状态: %s\n", state)
	if strings.ToLower(state) == "notready" {
		return false, nil
	}
	return true, nil
}

// 修改特定组件Managed值
func modifyComponentManaged(componentName string, managed bool, NPUOperatorCR unstructured.Unstructured,
	dynamicClient dynamic.Interface, ctx context.Context, gvr schema.GroupVersionResource) (unstructured.Unstructured, error) {
	// 记录NPUOperatorCR的初始状态
	initialNPUOperatorCR := NPUOperatorCR.DeepCopy()

	// 修改 Spec.componentName.Managed 为 false
	spec, found, err := unstructured.NestedMap(NPUOperatorCR.Object, "spec")
	if err != nil {
		return *initialNPUOperatorCR, fmt.Errorf("获取Spec字段失败：: %w", err)
	}
	if !found {
		return *initialNPUOperatorCR, fmt.Errorf("未找到Spec字段")
	}

	// 获取 componentName 字段
	mindioacp, found := spec[componentName].(map[string]interface{})
	if !found {
		return *initialNPUOperatorCR, fmt.Errorf("未找到%s字段", componentName)
	}

	// 设置 Managed
	mindioacp["managed"] = managed
	spec[componentName] = mindioacp

	// 设置修改后的 spec
	err = unstructured.SetNestedMap(NPUOperatorCR.Object, spec, "spec")
	if err != nil {
		return *initialNPUOperatorCR, fmt.Errorf("设置修改后的 Spec 失败: %w", err)
	}

	// 更新资源
	_, err = dynamicClient.Resource(gvr).Update(ctx, &NPUOperatorCR, metav1.UpdateOptions{})
	if err != nil {
		return *initialNPUOperatorCR, fmt.Errorf("更新%s组件的Managed失败: %w", componentName, err)
	}

	GinkgoWriter.Printf("成功更新%s组件的Managed子字段\n", componentName)
	return NPUOperatorCR, nil
}

// 获取组件状态
func getComponentState(componentName string, NPUOperatorCR unstructured.Unstructured) (string, string, error) {
	// 获取 Status 字段
	status, found, err := unstructured.NestedMap(NPUOperatorCR.Object, "status")
	if err != nil {
		return "", "", fmt.Errorf("获取Status字段失败: %w", err)
	}
	if !found {
		return "", "", fmt.Errorf("未找到Status字段")
	}

	// 获取 Component Statuses 数组
	componentStatuses, found, err := unstructured.NestedSlice(status, "componentStatuses")
	if err != nil {
		return "", "", fmt.Errorf("获取Component Statuses失败: %w", err)
	}
	if !found {
		return "", "", fmt.Errorf("未找到Component Statuses")
	}

	// 遍历 Component Statuses 查找匹配的组件
	for _, item := range componentStatuses {
		component, ok := item.(map[string]interface{})
		if !ok {
			continue // 跳过无效的条目
		}

		// 获取 Name 字段
		name, found, err := unstructured.NestedString(component, "name")
		if err != nil || !found {
			continue // 跳过没有 Name 字段的条目
		}

		// 检查 Name 是否包含 componentName
		if strings.Contains(name, componentName) {
			// 获取 State 字段
			state, found, err := unstructured.NestedMap(component, "state")
			if err != nil || !found {
				return "", "", fmt.Errorf("找到组件但未成功获取State字段: %s", name)
			}

			// 获取 Reason 字段
			reason, found, err := unstructured.NestedString(state, "reason")
			if err != nil || !found {
				return "", "", fmt.Errorf("找到组件但未成功获取State.Reason字段: %s", name)
			}

			// 获取 Type 字段
			currentType, found, err := unstructured.NestedString(state, "type")
			if err != nil || !found {
				return "", "", fmt.Errorf("找到组件但未成功获取State.Type字段: %s", name)
			}

			return reason, currentType, nil
		}
	}

	return "", "", fmt.Errorf("未找到包含 '%s' 的组件", componentName)
}
